package com.fjeld.middleware.activemq;


import javax.jms.*;

import com.fjeld.middleware.constants.PropertiesManager;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;

import java.io.Serializable;

import com.fjeld.server.store.reader.XMLKit;

/**
 * Generic topic listener, unserialising topics of the give topic.
 */
public class TopicListener implements MessageListener
{
    private static final Logger LOGGER = Logger.getLogger(TopicListener.class);

    private Session session;
    private Connection connection;
    protected Topic topic;
    protected String topicName, listenerName = "Default", logPrefix;

    public TopicListener(String stringTopic)
    {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BrokerEnum.BrokerURL.getValue());

        try
        {
            connection = factory.createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic(stringTopic);
            topicName = topic.getTopicName();
            logPrefix = "[" + listenerName + " on " + topicName + "] ";

            MessageConsumer consumer = session.createConsumer(topic);
            consumer.setMessageListener(this);

            connection.start();
            LOGGER.info(logPrefix + "Listening for messages.");

        } catch (JMSException e)
        {
            LOGGER.error("Exception setting up listener:", e);
        }
    }

    public TopicListener(String topicString, String listenerName)
    {
        // Should be in the form of server.service (e.g. ContentStoreServer.HeartbeatPublisher
        this.listenerName = listenerName;

        String hostName = PropertiesManager.getProperties().getProperty(listenerName + ".HostName", "localhost");
        String hostPort = PropertiesManager.getProperties().getProperty(listenerName + ".HostPort", "6161");
        String brokerURL = "tcp://" + hostName + ":" + hostPort;
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);

        try
        {
            connection = factory.createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic(topicString);
            topicName = topic.getTopicName();
            logPrefix = "[" + listenerName + " on " + topicName + "] ";

            MessageConsumer consumer = session.createConsumer(topic);
            consumer.setMessageListener(this);

            LOGGER.info(logPrefix + "TopicPublisher has been created");
        } catch (JMSException e)
        {
            LOGGER.error("Exception setting up listener:", e);
        }
    }

    public void startConnection()
    {
        try
        {
            LOGGER.debug(logPrefix + "Starting listener transport...");
            connection.start();
            LOGGER.debug(logPrefix + "Started listener transport.");
        } catch (JMSException e)
        {
            LOGGER.fatal(logPrefix + "Could not start listener.", e);
            System.exit(1);
        }


    }

    public void onMessage(Message message)
    {
        ObjectMessage objMsg = (ObjectMessage) message;
        try
        {
            Serializable msg = objMsg.getObject();
            //LOGGER.debug(logPrefix + "Received:\n" + XMLKit.xmlObjToString(msg));
        } catch (JMSException e)
        {
            LOGGER.error(logPrefix + "Could not unmarshal message:", e);
        }
    }
}
